package com.tandashi.springbootstudy01.sensor;

import com.tandashi.springbootstudy01.sensor.receiver.Receiver;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener;
import org.apache.kafka.common.utils.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Function;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static org.apache.commons.lang3.time.DateUtils.MILLIS_PER_SECOND;

@SuppressWarnings({"unused", "MismatchedQueryAndUpdateOfCollection"})
@Component
@Slf4j
public class SensorMessageConsumer {

    @Scheduled(initialDelayString = "${schedule.sensor.initialDelay:30000}", fixedDelayString = "${schedule.sensor.fixedDelay:10}")
    public void process() {
        try {
            if (!open) {
                return;
            }
            consume();
        } catch (Exception e) {
            log.error("consume sensor message error", e);
        }
    }

    private void consume() throws InterruptedException {
        ConsumerRecords<String, Bytes> records = poll();

        CountDownLatch doneSignal = new CountDownLatch(records.count());
        for (ConsumerRecord<String, Bytes> record : records) {
            executor.submit(() -> consume(record, doneSignal));
        }

        doneSignal.await();
        commit();
    }

    private void consume(ConsumerRecord<String, Bytes> record, CountDownLatch doneSignal) {
        try {
            log.warn("topic:{} partition:{} timestamp:{} key:{}", record.topic(), record.partition(), record.timestamp(), record.key());

            String key = key(record);
            Receiver receiver = findReceiver(key);

            if (null != receiver) {
                receiver.receive(record.value());
            } else {
                log.error("no receiver found for topic {} key {}", record.topic(), record.key());
            }

        } finally {
            doneSignal.countDown();
        }
    }

    private Receiver findReceiver(String key) {
        return receiverMap.get(key);
    }


    private ConsumerRecords<String, Bytes> poll() {
        return consumer.poll(MILLIS_PER_SECOND);
    }


    @PostConstruct
    private void postConstruct() {
        if (!open) {
            return;
        }

        receiverMap = receiverList.stream().collect(Collectors.toMap(SensorMessageConsumer::key, Function.identity()));

        subscribe();
    }

    private void subscribe() {
        consumer.subscribe(Pattern.compile(sensorTopics), new NoOpConsumerRebalanceListener());
    }

    private void unsubscribe() {
        consumer.unsubscribe();
    }

    private void commit() {
        consumer.commitSync();
    }

    private static String key(Receiver receiver) {
        if (null == receiver.tag()) {
            return receiver.topic();
        }

        return String.format("%s-%s", receiver.topic(), receiver.tag());
    }

    private static String key(ConsumerRecord<String, Bytes> record) {
        if (null == record.key() || record.topic().startsWith("act-")) {
            return record.topic();
        }
        return String.format("%s-%s", record.topic(), record.key());
    }

    @PreDestroy
    private void preDestory() {
        unsubscribe();
    }

    @Autowired
    private KafkaConsumer<String, Bytes> consumer;
    @Autowired
    private ScheduledExecutorService executor;


    @Autowired
    private List<Receiver> receiverList;
    private Map<String, Receiver> receiverMap;

    @Value("sensor.topics.pattern:wx-product-vw,wx-visit-callback|act-.*")
    private String sensorTopics;
    @Value("${schedule.sensor.open:true}")
    private boolean open;
}
